package service

import (
	"context"
	"fmt"
	"ginchat/models/mysqlModel"
	"ginchat/sql/userSql"
	"ginchat/utils/redisInit"
	"github.com/goccy/go-json"
	"github.com/gorilla/websocket"
	"github.com/redis/go-redis/v9"
	"github.com/spf13/viper"
	"gopkg.in/fatih/set.v0"
	"net"
	"net/http"
	"strconv"
	"sync"
	"time"
)

type Node struct {
	Conn *websocket.Conn
	Addr          string          //客户端地址
	FirstTime     uint64          //首次连接时间
	HeartbeatTime uint64          //心跳时间
	LoginTime     uint64          //登录时间
	DataQueue chan []byte
	GroupeSets set.Interface
}
//更新用户心跳
func (node *Node) Heartbeat(currentTime uint64) {
	node.HeartbeatTime = currentTime
	return
}
//用户心跳是否超时
func (node *Node) IsHeartbeatTimeOut(currentTime uint64) (timeout bool) {
	if node.HeartbeatTime+viper.GetUint64("timeout.HeartbeatMaxTime") <= currentTime {
		fmt.Println("心跳超时。。。自动下线", node)
		timeout = true
	}
	return
}
//清理超时连接
func CleanConnection(param interface{}) (result bool) {
	result = true
	defer func() {
		var s any
		if r := recover(); r != s {
			fmt.Println("cleanConnection err", r)
		}
	}()
	//fmt.Println("定时任务,清理超时连接 ", param)
	//node.IsHeartbeatTimeOut()
	currentTime := uint64(time.Now().Unix())
	for i := range clientMap {
		node := clientMap[i]
		if node.IsHeartbeatTimeOut(currentTime) {
			fmt.Println("心跳超时..... 关闭连接：", node)
			node.Conn.Close()
		}
	}
	return result
}

//存储客户端信息
var clientMap map[int]*Node = make(map[int]*Node,0)
//读写锁
var rwLocker sync.RWMutex

//需要：发送者ID 接收者ID 消息类型 发送的类型 消息内容
func Chat(writer http.ResponseWriter,request *http.Request,userId int){
	//2.校验token
	isvalid := true //checkToke() 待。。。。。。
	conn,err := (&websocket.Upgrader{
		CheckOrigin: func(r *http.Request) bool {
			return isvalid
		},
	}).Upgrade(writer,request,nil)
	if err != nil{
		fmt.Println("websocket.Upgrade  failed:",err)
		return
	}

	//3.获取conn
	currentTime := uint64(time.Now().Unix())
	node := &Node{
		Conn:       conn,
		Addr:          conn.RemoteAddr().String(), //客户端地址
		HeartbeatTime: currentTime,                //心跳时间
		LoginTime:     currentTime,                //登录时间
		DataQueue:  make(chan []byte,1024),
		GroupeSets: set.New(set.ThreadSafe),
	}
	//4.用户关系
	//5.userid 跟 node绑定 并加锁
	rwLocker.Lock()
	clientMap[int(userId)] = node
	rwLocker.Unlock()

	//6.完成发送逻辑
	go sendProc(node)
	//7.完成接受逻辑
	go recvProc(node)
	//7.加入在线用户到缓存
	SetUserOnlineInfo("online_"+string(userId), []byte(node.Addr), time.Duration(viper.GetInt("timeout.RedisOnlineTime"))*time.Hour)
	//sendMsg(int(userId),[]byte("欢迎进入聊天系统"))
}
/**
设置在线用户到redis缓存
**/
func SetUserOnlineInfo(key string, val []byte, timeTTL time.Duration) {
	ctx := context.Background()
	re.Set(ctx, key, val, timeTTL)
}
func sendProc(node *Node) {
	for{
		select{
		case data := <- node.DataQueue:
			err := node.Conn.WriteMessage(1,[]byte(data))
			if err != nil{
				fmt.Println("node.Conn.WriteMessage failed:",err)
				return
			}
		}
	}
}
func recvProc(node *Node){//接受消息并且处理消息
	for{
		//阻塞等待接受数据
		_,data,err := node.Conn.ReadMessage()
		if err != nil{
			fmt.Println("node.Conn.ReadMessage failed:",err)
			return
		}
		//将收到的数据 ---> 结构体
		msg := &mysqlModel.Message{}
		err = json.Unmarshal(data,msg)
		if err != nil{
			fmt.Println(err)
			return
		}
		if msg.Type == 3 {
			currentTime := uint64(time.Now().Unix())
			node.Heartbeat(currentTime)
		} else {
			dispatch(data)
			broadMsg(data) //todo 将消息广播到局域网
			fmt.Println("[ws] recvProc <<<<< ", string(data))
		}
	}
}

//后端调度逻辑以及后端调度函数实现
func dispatch(data []byte){
	msg := mysqlModel.Message{}
	msg.CreateTime = uint64(time.Now().Unix())
	//将数据映射到结构体里面
	err := json.Unmarshal(data, &msg)
	if err != nil {
		fmt.Println(err)
		return
	}
	switch msg.Type {
	case 1: //私信
		fmt.Println("dispatch  data :", string(data))
		sendMsg(msg.TargetId, data)
	case 2: //群发
		sendGroupMsg(int64(msg.TargetId), data) //发送的群ID ，消息内容
		// case 4: // 心跳
		// 	node.Heartbeat()
		//case 4:
		//
	}
}
func sendGroupMsg(targetId int64, msg []byte) {
	fmt.Println("开始群发消息")
	userIds := userSql.SearchUserByGroupId(uint(targetId))
	for i := 0; i < len(userIds); i++ {
		//排除给自己的
		if targetId != int64(userIds[i]) {
			sendMsg(int(userIds[i]), msg)
		}

	}
}

var re = redisInit.Get()
//sendMsg发送消息的整个逻辑
//获取接受者的conn(上锁读取) ---> 将msg转化为结构体 ---> 获取消息发送者id 接收者id 并且赋值消息的创建时间
//查看用户是否过期：过期就不发送消息 ---> 将发送的消息存储在redis中(为了获取历史消息)
func sendMsg(userId int,msg []byte){
	//获取私信目标的node
	rwLocker.RLock()
	node,ok := clientMap[userId]
	rwLocker.RUnlock()

	//将消息格式化
	jsonMsg := mysqlModel.Message{}
	json.Unmarshal(msg, &jsonMsg)
	ctx := context.Background()
	//targetIdStr：消息目标  userIdStr:消息发送者  以及消息的创建时间
	targetIdStr := strconv.Itoa(int(userId))
	userIdStr := strconv.Itoa(int(jsonMsg.FormId))
	jsonMsg.CreateTime = uint64(time.Now().Unix())

	//判断发送者是否过期了
	r,err := re.Get(ctx, "online_"+userIdStr).Result()
	if err != nil {
		fmt.Println(err)
	}
	if r != "" {
		if ok {
			fmt.Println("sendMsg >>> userID: ", userId, "  msg:", string(msg))
			node.DataQueue <- msg
		}
	}

	var key string
	//如果接受者目标的Id大于发送者的id ---》使key值唯一
	if userId > jsonMsg.FormId {
		key = "msg_" + userIdStr + "_" + targetIdStr
	} else {
		key = "msg_" + targetIdStr + "_" + userIdStr
	}
	//将key(有序集合)的所有元素按照score值从大小排列返回
	res, err := re.ZRevRange(ctx, key, 0, -1).Result()
	if err != nil {
		fmt.Println(err)
	}
	//cap ---> 获取消息的个数
	score := float64(cap(res)) + 1
	//将最新的消息存储在key里面---》数值对应它是第几条数据
	ress, e := re.ZAdd(ctx, key, redis.Z{score, msg}).Result() //jsonMsg
	//res, e := utils.Red.Do(ctx, "zadd", key, 1, jsonMsg).Result() //备用 后续拓展 记录完整msg
	if e != nil {
		fmt.Println(e)
	}
	fmt.Println(ress)
}

//获取缓存里面的消息
func RedisMsg(userIdA int64, userIdB int64, start int64, end int64, isRev bool) []string {
	ctx := context.Background()
	userIdStr := strconv.Itoa(int(userIdA))
	targetIdStr := strconv.Itoa(int(userIdB))
	var key string
	if userIdA > userIdB {
		key = "msg_" + targetIdStr + "_" + userIdStr
	} else {
		key = "msg_" + userIdStr + "_" + targetIdStr
	}
	var rels []string
	var err error
	if isRev {
		rels, err = re.ZRange(ctx, key, start, end).Result()
	} else {
		rels, err = re.ZRevRange(ctx, key, start, end).Result()
	}
	if err != nil {
		fmt.Println(err) //没有找到
	}
	// 发送推送消息
	/**
	// 后台通过websoket 推送消息
	for _, val := range rels {
		fmt.Println("sendMsg >>> userID: ", userIdA, "  msg:", val)
		node.DataQueue <- []byte(val)
	}**/
	return rels
}

func init() {
	go upSendProc()
	go upRecvProc()
	fmt.Println("init goroutine ")
}
var upsendChan chan []byte = make(chan []byte)
func broadMsg(data []byte){
	upsendChan <- data
}
//完成udp数据发送协程 ---》 192.16f.61.1
func upSendProc(){
	conn,err := net.DialUDP("udp",nil,&net.UDPAddr{
		IP: net.IPv4(192,168,0,255),
		Port: 3000,
	})
	if err != nil{
		fmt.Println("net.DialUDP failed:",err)
		return
	}
	defer conn.Close()

	for{
		select {
		case data := <- upsendChan:
			fmt.Println("upSendProc >>>>>>",string(data))
			_,err := conn.Write(data)
			if err != nil{
				fmt.Println("conn.Write failed:",err)
				return
			}
		}
	}
}
//完成udp数据接收协程
func upRecvProc(){
	conn,err := net.ListenUDP("udp",&net.UDPAddr{
		IP: net.IPv4zero,
		Port: 3000,
	})
	if err != nil{
		fmt.Println("net.ListenUDP failed:",err)
		return
	}
	defer conn.Close()

	for{
		buf := make([]byte,1024*6)
		n,err := conn.Read(buf)
		if err != nil{
			fmt.Println("conn.Read falied:",err)
			return
		}
		fmt.Println("upRecvProc <<<<<<",string(buf[:n]))
		dispatch(buf[:n])
	}
}


